package io.zbus.mq;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.zbus.mq.model.MessageQueue;
import io.zbus.mq.model.Subscription;

public class SubscriptionManager {  
	private static final Logger logger = LoggerFactory.getLogger(SubscriptionManager.class); 
	private Map<String, Subscription> subscriptionMap = new ConcurrentHashMap<>();
	private Map<String, List<Subscription>> channel2Subscription = new ConcurrentHashMap<>();
	
	private MessageQueueManager messageQueueManager;
	
	public SubscriptionManager(MessageQueueManager messageQueueManager) {
		this.messageQueueManager = messageQueueManager;
	}
	
	public synchronized Subscription get(String clientId) {
	    return this.get(clientId, "", "");
	}
    public synchronized Subscription get(String clientId, String mq, String channel) {
	    String subKey = Subkey(clientId, mq, channel);
		return subscriptionMap.get(subKey);
	}
	
	public synchronized void add(Subscription subscription) {
        String subKey = Subkey(subscription.clientId, subscription.mq, subscription.channel);
        if (subscriptionMap.containsKey(subKey))
        {
            return;
        }
		logger.info("add subscriber {}.", subKey);

        subscriptionMap.put(subKey, subscription);

		String channelKey = key(subscription.mq, subscription.channel);
        List<Subscription> subs = channel2Subscription.get(channelKey);
		if(subs == null) {
			subs = new ArrayList<>();
			channel2Subscription.put(channelKey, subs);
		}
		subs.add(subscription);
	}
	
	
	public synchronized List<Subscription> getSubscriptionList(String mq, String channel){
		String channelKey = key(mq, channel);
		return channel2Subscription.get(channelKey);
	}
	
	public synchronized void removeByClientId(String clientId) {
        Iterator<Entry<String, Subscription>> iter = subscriptionMap.entrySet().iterator();
        while(iter.hasNext()) {
            Entry<String, Subscription> entry = iter.next();
            if(clientId.equals(entry.getValue().clientId)) {
                iter.remove();
				MessageQueue mq = messageQueueManager.get(entry.getValue().mq);
				if(mq != null) {
                    Integer mask = mq.getMask();
                    if (mask != null && (Protocol.MASK_DELETE_ON_EXIT & mask) != 0) {
                        try {
                            messageQueueManager.removeQueue(mq.name(), null);
                        } catch (IOException e) {
                            logger.error(e.getMessage(), e);
                        }
                    }
                }
            }
        }

        Iterator<Entry<String, List<Subscription>>> it = channel2Subscription.entrySet().iterator();
        while(it.hasNext()) {
            Entry<String, List<Subscription>> entry = it.next();
            for(Subscription sub : entry.getValue()) {
                if(clientId.equals(sub.clientId)) {
                    it.remove();
					MessageQueue mq = messageQueueManager.get(sub.mq);
					if(mq != null) {
                        Integer mask = mq.getMask();
                        if (mask != null && (Protocol.MASK_DELETE_ON_EXIT & mask) != 0) {
                            try {
                                messageQueueManager.removeQueue(mq.name(), null);
                            } catch (IOException e) {
                                logger.error(e.getMessage(), e);
                            }
                        }
                        break;
                    }
                }
            }
        }
	}
	
	public synchronized void removeByChannel(String mq, String channel) {
		String channelKey = key(mq, channel);
		List<Subscription> subs = channel2Subscription.remove(channelKey);
		if(subs == null) return;
		
		for(Subscription sub : subs) {
		    String subKey = Subkey(sub.clientId, sub.mq, sub.channel);
		    if (!subscriptionMap.containsKey(subKey))
            {
                continue;
            }
			subscriptionMap.remove(subKey);
		}
	}
	
	
	public synchronized void removeByMq(String mq) { 
		Iterator<Entry<String, Subscription>> iter = subscriptionMap.entrySet().iterator();
		while(iter.hasNext()) {
			Entry<String, Subscription> entry = iter.next(); 
			if(mq.equals(entry.getValue().mq)) {
				iter.remove();
			}
		}
		
		Iterator<Entry<String, List<Subscription>>> it = channel2Subscription.entrySet().iterator();
		while(it.hasNext()) {
			Entry<String, List<Subscription>> entry = it.next();
			for(Subscription sub : entry.getValue()) {
				if(mq.equals(sub.mq)) {
					it.remove();
					break;
				}
			}
		}
	}
	
	private String key(String mq, String channel) {
		if(mq == null) mq = "";
		if(channel == null) channel = "";
		String channelKey = mq + "." + channel;
		return channelKey;
	}

    private String Subkey(String clientId, String mq, String channel) {
        if(clientId == null) clientId = "";
        if(mq == null) mq = "";
        if(channel == null) channel = "";
        String subKey = clientId + "." + mq + "." + channel;
        return subKey;
    }
}
